package com.zch.client.spring.mqtt.listener;

import com.alibaba.fastjson.JSON;
import com.zch.client.spring.mqtt.dto.MqttConsumerMessage;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttMessage;

/**
 * 抽象的消息订阅监听器
 */
@Slf4j
public abstract class AbstractMqttConsumerListener implements IMqttMessageListener {
    // messageArrived 抛出异常时关闭连接，是客户端库为保障系统稳定性而设计的保护机制
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        MqttConsumerMessage consumerMessage = new MqttConsumerMessage();
        consumerMessage.setMqttMessage(mqttMessage);
        consumerMessage.setTopic(topic);
         try{
           onMessage(consumerMessage);
         }catch (Throwable e){
             log.error("类:{} messageArrived 异常 topic:{},message:{}",this.getClass().getSimpleName(),topic,
                     JSON.toJSONString(consumerMessage));
             consumerMessage.setError(e);
             onError(consumerMessage,e);
         }
    }

    public abstract void onMessage(MqttConsumerMessage consumerMessage);
    //消费处理异常的处理
    public abstract void onError(MqttConsumerMessage consumerMessage,Throwable e);
}
